<?php
namespace Zeke\Io\Loop;

use Zeke\Zee as zz;
use Zeke\Io\Loop\Timer\Timer;
use Zeke\Io\Loop\Timer\Timers;
use Zeke\Io\Loop\Tick\FutureTickQueue;
use Zeke\Contracts\Io\Loop\LoopContract;
use Zeke\Contracts\Io\Loop\Timer\TimerContract;


class Select implements LoopContract
{


    /**
     * 微秒
     * @var int
     */
    const MICROSECONDS_PER_SECOND = 1000000;



    private $futureTickQueue;
    private $timers;
    private $readStreams = array();
    private $readListeners = array();
    private $writeStreams = array();
    private $writeListeners = array();
    private $running;
    private $pcntl = false;
    private $signals;


    /**
     *
     * Select constructor.
     */
    public function __construct()
    {
        $this->futureTickQueue  = new FutureTickQueue();
        $this->timers           = new Timers();
        $this->signals          = new SignalsHandler();
        $this->pcntl            = \extension_loaded('pcntl');
    }


    /**
     * 添加读stream
     *
     * @param $stream
     * @param $listener
     * @author liu.bin 2018/8/2 16:25
     */
    public function addReadStream($stream, $listener)
    {
        $fd = zz\intFd($stream);
        if (!isset($this->readStreams[$fd])) {
            $this->readStreams[$fd]     = $stream;
            $this->readListeners[$fd]   = $listener;
        }
    }


    /**
     * 添加写stream
     *
     * @param $stream
     * @param $listener
     * @author liu.bin 2018/8/2 18:52
     */
    public function addWriteStream($stream, $listener)
    {
        $fd = zz\intFd($stream);
        if (!isset($this->writeStreams[$fd])) {
            $this->writeStreams[$fd] = $stream;
            $this->writeListeners[$fd] = $listener;
        }
    }



    /**
     * 移除读stream
     *
     */
    public function removeReadStream($stream)
    {
        $fd = zz\intFd($stream);
        unset(
            $this->readStreams[$fd],
            $this->readListeners[$fd]
        );
    }


    /**
     * 移除写stream
     *
     * @param $stream
     * @author liu.bin 2018/8/2 18:53
     */
    public function removeWriteStream($stream)
    {
        $fd = zz\intFd($stream);
        unset(
            $this->writeStreams[$fd],
            $this->writeListeners[$fd]
        );
    }


    /**
     * 事件循环
     *
     * @author liu.bin 2018/8/2 18:53
     */
    public function run()
    {
        $this->running = true;


        while ($this->running) {

            //轮询执行一遍所有的队列任务
            $this->futureTickQueue->tick();
            //轮询执行一遍所有的定时器任务
            $this->timers->tick();


            /**
             * 1: futureTickQueue任务执行完即出列被注销掉
             * 2: running 停止，futureTickQueue任务不为空（futureTickQueue任务可能在读写回调用注册）
             *      仍然需要开启IOLoop非阻塞返回执行futureTickQueue任务
             */
            if (!$this->running || !$this->futureTickQueue->isEmpty()) {

                // Future-tick queue has pending callbacks ...
                $timeout = 0;


                //getFirst() 最先执行的定时器任务发生的时间点
                //$scheduledAt 为时间事件注册的当前时间＋间隔事件
            } elseif ($scheduledAt = $this->timers->getFirst()) {


                // There is a pending timer, only block until it is due ...
                //$this->timers->getTime();当前时间毫秒
                $timeout = $scheduledAt - $this->timers->getTime();
                if ($timeout < 0) {

                    $timeout = 0;
                } else {

                    // Convert float seconds to int microseconds.
                    // Ensure we do not exceed maximum integer size, which may
                    // cause the loop to tick once every ~35min on 32bit systems.
                    //秒－>毫秒:支持小数的秒
                    $timeout *= self::MICROSECONDS_PER_SECOND;
                    $timeout = $timeout > \PHP_INT_MAX ? \PHP_INT_MAX : (int)$timeout;
                }
                // The only possible event is stream or signal activity, so wait forever ...
            } elseif ($this->readStreams || $this->writeStreams || !$this->signals->isEmpty()) {

                //null 表示永不过期，会一直等待IO事件发生才返回
                $timeout = null;
                // There's nothing left to do ...
            } else {

                break;
            }



            //等待/检测 stream 的活动情况，有定时器到期也会激活
            $this->waitForStreamActivity($timeout);
        }


    }


    /**
     * 事件停止
     *
     * @author liu.bin 2018/8/2 18:53
     */
    public function stop()
    {
        $this->running = false;
    }



    /**
     * 等待/检测 stream 的活动情况，有定时器到期也会激活
     * Wait/check for stream activity, or until the next timer is due.
     *
     * @param integer|null $timeout Activity timeout in microseconds, or null to wait forever.
     */
    private function waitForStreamActivity($timeout)
    {
        $read       = $this->readStreams;
        $write      = $this->writeStreams;
        $available  = $this->streamSelect($read, $write, $timeout);


        //调用已注册的信号回调函数
        if ($this->pcntl) {
            \pcntl_signal_dispatch();
        }


        if (false === $available) {
            // if a system call has been interrupted,
            // we cannot rely on it's outcome
            return;
        }


        //可读事件回调
        foreach ($read as $stream) {
            $key = (int) $stream;
            if (isset($this->readListeners[$key])) {
                \call_user_func($this->readListeners[$key], $stream);
            }
        }


        //可写事件回调
        foreach ($write as $stream) {
            $key = (int) $stream;
            if (isset($this->writeListeners[$key])) {
                \call_user_func($this->writeListeners[$key], $stream);
            }
        }
    }



    /**
     * Emulate a stream_select() implementation that does not break when passed
     * empty stream arrays.
     *
     * @param array        &$read   An array of read streams to select upon.
     * @param array        &$write  An array of write streams to select upon.
     * @param integer|null $timeout Activity timeout in microseconds, or null to wait forever.
     *
     * @return integer|false The total number of streams that are ready for read/write.
     * Can return false if stream_select() is interrupted by a signal.
     */
    private function streamSelect(array &$read, array &$write, $timeout)
    {
        if ($read || $write) {
            $except = null;
            // suppress warnings that occur, when stream_select is interrupted by a signal
            return @\stream_select($read, $write, $except, $timeout === null ? null : 0, $timeout);
        }
        $timeout && \usleep($timeout);
        return 0;
    }


    /**
     * 定时一次性任务
     *
     * @param float|int $interval
     * @param callable $callback
     * @return TimerContract|Timer
     *
     * @author <liu.bin@ruaby.com> 2018-08-19 21:49
     */
    public function addTimer($interval, $callback)
    {
        $timer = new Timer($interval, $callback, false);

        $this->timers->add($timer);

        return $timer;
    }


    /**
     * 定时循环任务
     *
     * @param float|int $interval
     * @param callable $callback
     * @return TimerContract|Timer
     *
     * @author <liu.bin@ruaby.com> 2018-08-19 21:48
     */
    public function addPeriodicTimer($interval, $callback)
    {
        $timer = new Timer($interval, $callback, true);

        $this->timers->add($timer);

        return $timer;
    }

    /**
     * Cancel a pending timer.
     *
     * See also [`addPeriodicTimer()`](#addperiodictimer) and [example #2](examples).
     *
     * Calling this method on a timer instance that has not been added to this
     * loop instance or on a timer that has already been cancelled has no effect.
     *
     * @param TimerContract $timer The timer to cancel.
     *
     * @return void
     */
    public function cancelTimer(TimerContract $timer)
    {
        $this->timers->cancel($timer);
    }

    /**
     * 每次IoLoop事件开始，会执行此队列全部任务
     *
     * @param callable $listener The callback to invoke.
     *
     * @return void
     *
     * 此方法属于AOP编程范畴，可以预先注册一些任务在Loop，在Loop启动的时候全部执行
     */
    public function futureTick($listener)
    {
        $this->futureTickQueue->add($listener);
    }

    /**
     * 同一个信号量可以注册多个回调处理函数
     * @param int $signal
     * @param callable $listener
     *
     * @throws \BadMethodCallException when signals aren't supported on this
     *     platform, for example when required extensions are missing.
     *
     * @return void
     */
    public function addSignal($signal, $listener)
    {
        if ($this->pcntl === false) {
            throw new \BadMethodCallException('Event loop feature "signals" isn\'t supported by the "StreamSelectLoop"');
        }

        //第一次注册
        $first = $this->signals->count($signal) === 0;
        $this->signals->add($signal, $listener);

        if ($first) {
            //第一次需要定义回调函数
            \pcntl_signal($signal, array($this->signals, 'call'));
        }
    }

    /**
     * Removes a previously added signal listener.
     *
     * ```php
     * $loop->removeSignal(SIGINT, $listener);
     * ```
     *
     * Any attempts to remove listeners that aren't registered will be ignored.
     *
     * @param int $signal
     * @param callable $listener
     *
     * @return void
     */
    public function removeSignal($signal, $listener)
    {
        if (!$this->signals->count($signal)) {
            return;
        }

        $this->signals->remove($signal, $listener);

        if ($this->signals->count($signal) === 0) {
            //这个符号表示恢复系统对信号的默认处理
            \pcntl_signal($signal, \SIG_DFL);
        }
    }
}